﻿using Confluent.Kafka;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;


namespace BCGPS.DataReceive.Service
{
    public class KafkaConfluentProducer
    {
        public void test()
        {
            var config = new ProducerConfig { BootstrapServers = ConfigurationManager.AppSettings["Kafka:BrokerList"]     };
        }

        public bool Produce(string broker, string topic, int partion, string message)
        {
            bool result = false;
            if (string.IsNullOrEmpty(broker) || string.IsNullOrWhiteSpace(broker) || broker.Length <= 0)
            {
                throw new ArgumentNullException("Kafka消息服务器地址不能为空");
            }

            if (string.IsNullOrEmpty(topic) || string.IsNullOrWhiteSpace(topic) || topic.Length <= 0)
            {
                throw new ArgumentNullException("消息所属的主题不能为空！");
            }

            if (string.IsNullOrEmpty(message) || string.IsNullOrWhiteSpace(message) || message.Length <= 0)
            {
                throw new ArgumentNullException("消息内容不能为空！");
            }

            var config = new Dictionary<string, object>
            {
                {"bootstrap.serviers",broker }
            };

            return true; ;
        }
    }
}
